package cn.itcast.connectors;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.Properties;

/**
 * @author KTL
 * @version V1.0
 * @Package cn.itcast.connectors
 * @date 2021/2/25 0025 9:07
 * @Copyright © 2015-04-29  One for each, and two for each
 *
 */
public class KafkaSinkDemo {

    public static void main(String[] args) throws Exception {
        //TODO -0 env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //TODO -1 source
        //设置kafka链接信息
        final Properties properties = new Properties();
        properties.setProperty("","");
        final FlinkKafkaConsumer<String> kafkasource = new FlinkKafkaConsumer<>("flink_kafka", new SimpleStringSchema(), properties);
        final DataStreamSource<String> kafkaDS = env.addSource(kafkasource);
        //TODO -2 transfromation
        final SingleOutputStreamOperator<String> etlDS = kafkaDS.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return value.contains("success");
            }
        });
        //TODO -3 sink
        final Properties propertiesProducer = new Properties();
        final FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>("flink_kafka_ropertie", new SimpleStringSchema(), propertiesProducer);
        etlDS.addSink(kafkaSink);
        //TODO -4 execute
        env.execute();
    }
}
/**
 * 准备主题
 * 启动控制台生产者
 * 启动程序
 */